package belf.migrate.plugin.sqlserver;

import belf.migrate.api.common.DatabaseIdentifier;
import belf.migrate.api.constants.SQLConstants;
import belf.migrate.api.exception.*;
import belf.migrate.api.model.CharacteristicInfo;
import belf.migrate.api.model.TableInfoModel;
import belf.migrate.api.table.schema.*;
import belf.migrate.api.util.CatalogUtils;
import belf.migrate.api.util.ExceptionUtils;
import belf.migrate.api.util.JdbcUrlUtil;
import belf.migrate.plugin.mysql.antlr.MySqlLexer;
import belf.migrate.plugin.mysql.antlr.MySqlParser;
import belf.migrate.plugin.sqlserver.antlr.TSqlLexer;
import belf.migrate.plugin.sqlserver.antlr.TSqlParser;
import belf.migrate.plugin.sqlserver.antlr.TSqlParserBaseListener;
import belf.migrate.plugin.sqlserver.listener.SqlServerCharacteristicListener;
import belf.migrate.plugin.sqlserver.listener.SqlServerFunctionListener;
import lombok.extern.slf4j.Slf4j;
import org.antlr.v4.runtime.CharStream;
import org.antlr.v4.runtime.CharStreams;
import org.antlr.v4.runtime.CommonTokenStream;
import org.antlr.v4.runtime.tree.ParseTree;
import org.antlr.v4.runtime.tree.ParseTreeWalker;
import org.apache.commons.lang3.StringUtils;

import java.sql.*;
import java.util.*;
import java.util.function.Predicate;

@Slf4j
public class SqlServerCatalog extends AbstractSinkCatalog {
    private static final String QUOTE = "\"";   //sqlserver还是可以用双引号做quote的

    private static final String[][] TSQL_BEGIN_END_BLOCK = new String[][]{
            {SQLConstants.BLOCK_BEGIN, SQLConstants.BLOCK_END}
    };

    static {
        SYS_DATABASES.addAll(Collections.unmodifiableList(
                Arrays.asList("guest", "INFORMATION_SCHEMA", "sys", "db_owner",
                        "db_accessadmin", "db_securityadmin", "db_ddladmin",
                        "db_backupoperator", "db_datareader", "db_datawriter",
                        "db_denydatareader", "db_denydatawriter")));
    }

    public SqlServerCatalog(String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(DatabaseIdentifier.SQLSERVER, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    public SqlServerCatalog(String catalogName, String username, String pwd, JdbcUrlUtil.UrlInfo urlInfo, String defaultSchema, boolean caseSensitive) {
        super(catalogName, username, pwd, urlInfo, defaultSchema, caseSensitive);
    }

    @Override
    public String getFullTableName(TablePath tablePath) {
        return tablePath.getFullNameWithQuoted();
    }

    @Override
    public String getListDatabaseSQL() {
        return """
                SELECT name AS DatabaseName
                FROM sys.databases
                ORDER BY name;
                """;
    }

    @Override
    public String getListTableSQL(String database, String schema) {
        return String.format("""
                SELECT t.name as TABLE_NAME
                FROM sys.tables t
                LEFT JOIN sys.schemas s ON t.schema_id = s.schema_id
                WHERE s.name='%s'
                ORDER BY t.name
                """, schema);
    }

    /**
     * Sql Server中，使用SELECT TOP 1 * FROM TABLE来检查表是否存在数据
     * @param tablePath
     * @return
     */
    public String getExistDataSql(TablePath tablePath) {
        if (caseSensitive()) {
            return String.format(
                    "SELECT TOP 1 * FROM %s.%s",
                    tablePath.getRightDatabaseName().toUpperCase(), tablePath.getTableName().toUpperCase());
        } else {
            return String.format(
                    "SELECT TOP 1 * FROM %s.%s",
                    tablePath.getRightDatabaseName(), tablePath.getTableName());
        }
    }

    /**
     * 不指定数据库名称，则数据库名称需要在jdbc的url里
     * @param schemaName 模式名称
     * @return
     */
    @Override
    public String getListViewsSQL(String schemaName) {
        return getListViewsSQLInfo(defaultDatabase, schemaName);
    }

    @Override
    public String getListViewsSQLInfo(String defaultDatabase, String schemaName) {
        return String.format("""
                SELECT 
                    v.name AS VIEW_NAME
                FROM 
                    sys.views v
                JOIN 
                    sys.schemas s ON v.schema_id = s.schema_id
                WHERE 
                    s.name = '%s'
                ORDER BY 
                    v.name
                """,
                schemaName);
    }

    @Override
    public String getListFunctionSQL(String schemaName) {
        return String.format("""
                SELECT
                    o.name AS FUNCTION_NAME
                FROM
                    sys.objects o,sys.schemas s
                WHERE
                    o.schema_id = s.schema_id
                    AND s.name='%s'
                    AND o.type IN ('FN', 'IF', 'TF')
                ORDER BY
                    o.name
                """, schemaName);
    }

    @Override
    public String getListTriggerSQL(String schemaName) {
        return String.format("""
                SELECT
                    t.name AS TRIGGER_NAME,
                    o.name AS TABLE_NAME,
                    s.name AS SCHEMA_NAME
                FROM
                    sys.triggers t
                JOIN
                    sys.objects o ON t.parent_id = o.object_id
                JOIN
                	sys.schemas s ON s.schema_id = o.schema_id
                WHERE
                    o.type IN ('U', 'V')  -- 'U': User tables, 'V': Views
                    AND s.name='%s'
                ORDER BY
                    t.name
                """, schemaName);
    }

    @Override
    public String getListProcedureSQL(String schemaName) {
        return String.format("""
                SELECT
                    p.name AS PROCEDURE_NAME
                FROM
                    sys.procedures p
                JOIN
                    sys.schemas s ON p.schema_id = s.schema_id
                WHERE
                    s.name = '%s'
                ORDER BY
                    p.name;
                """, schemaName);
    }

    @Override
    public String getListTypeSQL(String schemaName) {
        return String.format("""
                SELECT
                    t.name AS TYPE_NAME,
                    s.name AS SCHEMA_NAME
                FROM
                    sys.types t
                JOIN
                    sys.schemas s ON t.schema_id = s.schema_id
                WHERE
                    s.name = '%s'
                ORDER BY
                     t.name
                """, schemaName);
    }

    @Override
    public String getListSynonymSQL(String schemaName) {
        return String.format("""
                SELECT
                    syn.name AS SYNONYM_NAME
                FROM
                    sys.synonyms syn
                JOIN
                    sys.schemas s ON syn.schema_id = s.schema_id
                WHERE
                    s.name = '%s'
                ORDER BY
                    syn.name;
                """, schemaName);
    }

    @Override
    protected void fillingColumnExtra(TablePath tablePath, List<ColumnMetaData> columns, Connection connection) {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();
        String sql = String.format("""
                SELECT
                	s.name AS SchemaName,
                    t.name AS TableName,
                    c.name AS COLUMN_NAME,
                    dc.definition AS COLUMN_DEFAULT
                FROM
                    sys.columns c
                JOIN
                    sys.tables t ON c.object_id = t.object_id
                JOIN
                	sys.schemas s ON s.schema_id = t.schema_id
                LEFT JOIN
                    sys.default_constraints dc ON c.default_object_id = dc.object_id
                WHERE
                	s.name = '%s'
                    AND t.name = '%s'
                ORDER BY
                    c.column_id;
                """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try (Statement statement = connection.createStatement()) {
            log.info("Query Column Extra Info: sql=\n{}", sql);
            try (ResultSet rs = statement.executeQuery(sql)) {
                while (rs.next()) {
                    String columnName = rs.getString("COLUMN_NAME");
                    String columnDefault = rs.getString("COLUMN_DEFAULT");

                    Optional<ColumnMetaData> possibleColumn = columns.stream().filter(new Predicate<ColumnMetaData>() {
                        @Override
                        public boolean test(ColumnMetaData columnMetaData) {
                            return columnMetaData.getColumnName().equalsIgnoreCase(columnName);
                        }
                    }).findFirst();
                    if (possibleColumn.isPresent()) {
                        ColumnMetaData columnMetaData = possibleColumn.get();
                        ColumnExtra columnExtra = new ColumnExtra(columnDefault, null, null, 0);
                        columnMetaData.setExtra(columnExtra);
                    }
                }
            }
        } catch (SQLException e) {
            log.error("Error on query sqlserver column extra information:\n{}", ExceptionUtils.toString(e));
        }
    }

    @Override
    public List<IndexKey> getIndexKeys(TablePath tablePath) throws TableNotExistException, CatalogException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        if (!tableExists(tablePath)) {
            throw new TableNotExistException(catalogName, tablePath);
        }

        String dbUrl = defaultUrl;

        // 查询指定表的索引，排除主键索引
        String sql = String.format("""
                SELECT
                	s.name AS SchemaName,
                    t.name AS TableName,
                    i.name AS IndexName,
                    i.index_id AS IndexID,
                    i.type_desc AS IndexType,
                    c.name AS ColumnName,
                    ic.key_ordinal AS KeyOrdinal,
                    ic.is_descending_key AS IsDescending,
                    i.is_unique AS IsUnique,
                    i.is_primary_key AS IsPrimaryKey,
                    i.is_unique_constraint AS IsUniqueConstraint
                FROM
                    sys.indexes i
                JOIN
                    sys.tables t ON i.object_id = t.object_id
                JOIN
                	sys.schemas s ON s.schema_id = t.schema_id    
                JOIN
                    sys.index_columns ic ON i.object_id = ic.object_id AND i.index_id = ic.index_id
                JOIN
                    sys.columns c ON ic.object_id = c.object_id AND ic.column_id = c.column_id
                WHERE
                    s.name = '%s'
                    AND t.name = '%s'
                    AND i.is_primary_key != 1
                ORDER BY
                    i.name, ic.key_ordinal
                """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            Connection conn = getConnection(dbUrl);
            DatabaseMetaData metaData = conn.getMetaData();
            List<IndexKey> defaultIndexKeys =  CatalogUtils.getIndexKeys(metaData, tablePath);
            Statement statement = conn.createStatement();

            // index name -> index
            Map<String, IndexKeyExt> indexKeyMap = new HashMap<>();
            try (ResultSet rs = statement.executeQuery(sql)) {
                log.debug("query index:{}", sql);
                while (rs.next()) {
                    boolean unique = rs.getInt("IsUnique") == 1?true:false;
                    String indexName = rs.getString("IndexName");
                    int sequenceInIndex = rs.getInt("KeyOrdinal");
                    String columnName = rs.getString("ColumnName");
                    String indexOrder = rs.getInt("IsDescending")==1?"DESC":"ASC";
                    String indexTypeName = rs.getString("IndexType");
                    List<IndexKeyExt.ColumnWithSequence> columns = new ArrayList<>();
                    columns.add(new IndexKeyExt.ColumnWithSequence(columnName, sequenceInIndex));
                    IndexKeyExt IndexKeyExt = new IndexKeyExt(
                            columns,
                            indexName,
                            unique,
                            indexOrder,
                            indexTypeName);
                    IndexKeyExt existOne = indexKeyMap.get(indexName);
                    if (existOne != null) {
                        existOne.getColumns().addAll(IndexKeyExt.getColumns());
                    } else {
                        indexKeyMap.put(indexName, IndexKeyExt);
                    }

                    log.debug("got index: {}", IndexKeyExt);
                }
            }

            Iterator<IndexKey> it = defaultIndexKeys.iterator();
            while (it.hasNext()) {
                IndexKey indexKey = it.next();
                IndexKeyExt IndexKeyExt = indexKeyMap.get(indexKey.getIndexName());
                if (null != IndexKeyExt) {
                    indexKey.setColumnNames(IndexKeyExt.getSortedColumnNames());
                    indexKey.setIndexTypeName(IndexKeyExt.getIndexTypeName());
                    if (IndexKeyExt.getIndexOrder() != null
                            && IndexKeyExt.getIndexOrder().equalsIgnoreCase("ASC")) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.ASC);
                    } else if (IndexKeyExt.getIndexOrder() != null
                            && IndexKeyExt.getIndexOrder().equalsIgnoreCase("DESC")) {
                        indexKey.setIndexOrder(IndexKey.IndexOrder.DESC);
                    }
                } else { // 查找索引不包含主键
                    it.remove();
                }
            }

            statement.close();
            return defaultIndexKeys;
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public List<UniqueKey> getUniqueKeys(TablePath tablePath) throws CatalogException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                SELECT
                	tc.table_name AS TABLE_NAME,
                	kcu.column_name AS COLUMN_NAME,
                    tc.constraint_name AS CONSTRAINT_NAME
                FROM
                    information_schema.table_constraints AS tc
                JOIN
                    information_schema.key_column_usage AS kcu
                    ON tc.constraint_name = kcu.constraint_name
                    AND tc.table_name = kcu.table_name
                WHERE
                	tc.constraint_type = 'UNIQUE'
                    AND tc.table_name = '%s'
                ORDER BY
                	kcu.ordinal_position
                """, tablePath.getTableName());

        Map<String, UniqueKey> uniqueKeyMap = new HashMap<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(Statement statement = connection.createStatement()) {
            log.info("Query Table Unique Constraint: sql=\n{}", sql);
            try (ResultSet rs = statement.executeQuery(sql)){
                while (rs.next()) {
                    String constraintName = rs.getString("CONSTRAINT_NAME");
                    String column = rs.getString("COLUMN_NAME");
                    if (uniqueKeyMap.containsKey(constraintName)) {
                        UniqueKey value = uniqueKeyMap.get(constraintName);
                        value.addColumn(column);
                    } else {
                        List<String> columns = new ArrayList<>();
                        columns.add(column);
                        UniqueKey uniqueKey = new UniqueKey(tablePath.getTableName(), constraintName, columns);
                        uniqueKeyMap.put(constraintName, uniqueKey);
                    }
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "SQL Server查询Unique约束失败：table=" + tablePath.getTableName(), e);
        }
        return uniqueKeyMap.values().parallelStream().toList();
    }

    @Override
    public List<CheckConstraint> getCheckConstraints(TablePath tablePath) throws CatalogException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                SELECT
                    s.name AS SchemaName,
                    obj.name AS TableName,
                    col.name AS ColumnName,
                    con.name AS ConstraintName,
                    con.definition AS CheckCondition
                FROM
                    sys.check_constraints con
                JOIN
                    sys.objects obj ON con.parent_object_id = obj.object_id
                JOIN
                	sys.schemas s ON s.schema_id = obj.schema_id
                JOIN
                    sys.columns col ON con.parent_object_id = col.object_id
                        AND con.parent_column_id = col.column_id
                WHERE
                	s.name = '%s'
                    AND obj.name = '%s'
                """,
                tablePath.getSchemaName(),
                tablePath.getTableName());
        List<CheckConstraint> checkConstraints = new ArrayList<>();
        String dbUrl = getUrlFromDatabaseName(tablePath.getRightDatabaseName());
        Connection connection = getConnection(dbUrl);

        try(Statement ps = connection.createStatement()) {
            log.info("Query Table Check Constraint: sql=\n{}", sql);
            try (ResultSet rs = ps.executeQuery(sql)){
                while (rs.next()) {
                    String constraintName = rs.getString("ConstraintName");
                    String condition = rs.getString("CheckCondition");
                    checkConstraints.add(new CheckConstraint(tablePath.getTableName(),
                            constraintName,
                            condition));
                }
            }
        } catch (SQLException e) {
            e.printStackTrace();
            throw new CatalogException(APIErrorCode.SQL_EXECUTE_ERROR, "SQL Server中查询Check约束失败：table=" + tablePath.getTableName(), e);
        }
        return checkConstraints;
    }

    @Override
    public String showCreateTable(TablePath tablePath) throws TableNotExistException {
        return "";
    }

    @Override
    public String showCreateView(TablePath tablePath) throws ViewNotExistException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                SELECT
                	OBJECT_DEFINITION(obj.object_id) AS VIEW_DDL
                FROM
                	sys.objects obj
                JOIN
                	sys.schemas s ON obj.schema_id = s.schema_id
                WHERE
                	s.name = '%s'
                	AND obj.type = 'V'
                	and obj.name ='%s'
                """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            return getDDL(sql, "VIEW_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create view: \n{}", sql);
            throw new ViewNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
    }

    @Override
    public String showCreateProcedure(TablePath tablePath) throws ProcedureNotExistException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                SELECT
                	OBJECT_DEFINITION(obj.object_id) AS PROCEDURE_DDL
                FROM
                	sys.objects obj
                JOIN
                	sys.schemas s ON obj.schema_id = s.schema_id
                WHERE
                	s.name = '%s'
                	AND obj.type = 'P'
                	and obj.name ='%s'
                """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            return getDDL(sql, "PROCEDURE_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create procedure: \n{}", sql);
            throw new ProcedureNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
    }

    @Override
    public String showCreateTrigger(TablePath tablePath) throws TriggerNotExistException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                        SELECT
                        	OBJECT_DEFINITION(obj.object_id) AS TRIGGER_DDL
                        FROM
                        	sys.objects obj
                        JOIN
                        	sys.schemas s ON obj.schema_id = s.schema_id
                        WHERE
                        	obj.type = 'TR'
                        	AND s.name = '%s'
                        	AND obj.name ='%s'
                        """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            return getDDL(sql, "TRIGGER_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create trigger: \n{}", sql);
            throw new TriggerNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
    }

    @Override
    public String showCreateFunc(TablePath tablePath) throws FunctionNotExistException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                        SELECT
                        	OBJECT_DEFINITION(obj.object_id) AS FUNCTION_DDL
                        FROM
                        	sys.objects obj
                        JOIN
                        	sys.schemas s ON obj.schema_id = s.schema_id
                        WHERE
                        	obj.type IN ('AF', 'FN', 'FS', 'FT', 'IF')
                        	AND s.name = '%s'
                        	AND obj.name ='%s'
                        """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            return getDDL(sql, "FUNCTION_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create function: \n{}", sql);
            throw new FunctionNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
    }

    @Override
    public String showCreateType(TablePath tablePath) throws TypeNotExistException {
        String database = StringUtils.isEmpty(tablePath.getDatabaseName())? defaultDatabase : tablePath.getDatabaseName();

        String sql = String.format("""
                        SELECT
                        	OBJECT_DEFINITION(obj.object_id) AS TYPE_DDL
                        FROM
                        	sys.objects obj
                        JOIN
                        	sys.schemas s ON obj.schema_id = s.schema_id
                        WHERE
                        	obj.type = 'TT'
                        	AND s.name = '%s'
                        	AND obj.name ='%s'
                        """,
                tablePath.getSchemaName(),
                tablePath.getTableName());

        try {
            return getDDL(sql, "TYPE_DDL");
        } catch (CatalogException e) {
            log.error("Error while show create type: \n{}", sql);
            throw new TypeNotExistException(DatabaseIdentifier.SQLSERVER, tablePath, e);
        }
    }

    @Override
    public List<String> listSchemasByDatabase(String databaseName) throws CatalogException {
        String listSchemaSQL = """
                SELECT
                	name
                FROM
                	sys.schemas
                """;
        try {
            return queryString(
                    defaultUrl,
                    listSchemaSQL,
                    rs -> {
                        String s = rs.getString(1);
                        return SYS_DATABASES.contains(s) ? null : s;
                    });
        } catch (Exception e) {
            log.error("listSchemas failed", e);
            throw new CatalogException(
                    String.format("Failed listing schema in database %s", databaseName), e);
        }
    }

    @Override
    protected String getColumnsSql(TablePath tablePath) {
        return "SELECT\n" +
                "        a.colorder ordinal_position,\n" +
                "        '"+ tablePath.getDatabaseName() + "' as catalog_name,\n" +
                "        s.name as schema_name,\n" +
                "        d.name table_name,\n" +
                "        a.name column_name,\n" +
                "        b.name data_type,\n" +
                "        b.name column_type,\n" +
                "        (case when a.isnullable=1 then 1 else 0 end) nullable,\n" +
                "        isnull(e.text,'') data_default,\n" +
                "        isnull(g.[value], '') AS column_comment,\n" +
                "        a.xtype column_key,\n" +
                "        (\n" +
                "        case\n" +
                "        when (\n" +
                "        SELECT\n" +
                "        count(*)\n" +
                "        FROM\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.sysobjects\n" +
                "        WHERE\n" +
                "        (name in (\n" +
                "        SELECT\n" +
                "        name\n" +
                "        FROM\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.sysindexes\n" +
                "        WHERE\n" +
                "        (id = a.id) AND (indid in\n" +
                "        (SELECT\n" +
                "        indid\n" +
                "        FROM\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.sysindexkeys\n" +
                "        WHERE\n" +
                "        (id = a.id) AND (colid in\n" +
                "        (SELECT\n" +
                "        colid\n" +
                "        FROM\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.syscolumns\n" +
                "        WHERE\n" +
                "        (id = a.id) AND (name = a.name)\n" +
                "        ))\n" +
                "        ))\n" +
                "        ))\n" +
                "        AND (xtype = 'PK')\n" +
                "        ) > 0 then 1\n" +
                "        else 0\n" +
                "        end) is_primary_key,\n" +
                "        (CASE WHEN i.object_id > 0 THEN 1 ELSE 0 END ) is_increment,\n" +
                "        a.length as data_length,\n" +
                "        a.prec as prec,\n" +
                "        isc.character_maximum_length as char_max_length,\n" +
                "        isnull(a.scale,0) as precision,\n" +
                "        ic.is_identity as is_identity,\n" +
                "        ic.seed_value as identity_start,\n" +
                "        ic.increment_value as identity_step\n" +
                "        FROM\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.syscolumns a\n" +
                "        left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.types b on a.xusertype=b.user_type_id\n" +
                "        inner join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.sysobjects d on a.id=d.id and d.xtype='U' and d.name != 'dtproperties'\n" +
                "        left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.syscomments e on a.cdefault=e.id\n" +
                "        left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.extended_properties g on a.id=g.major_id AND a.colid=g.minor_id\n" +
                "        and g.class = 1 and g.name = 'MS_Description'  left join    ["+ tablePath.getDatabaseName() + "].sys.extended_properties f on d.id=f.class and f.minor_id=0  left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.identity_columns ic on ic.object_id = a.id and a.colid = ic.column_id\n" +
                "        inner join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.objects h on d.id = h.object_id\n" +
                "        inner join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.schemas s on h.schema_id = s.schema_id\n" +
                "        left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].sys.identity_columns i on i.object_id=h.object_id and i.column_id= a.colid\n" +
                "        left join\n" +
                "        ["+ tablePath.getDatabaseName() + "].information_schema.columns isc on isc.table_schema = s.name\n" +
                "        and isc.table_name = d.name and isc.column_name = a.name\n" +
                "        where\n" +
                "        1=1  and s.name = '" + tablePath.getSchemaName() + "' and d.name = '"+ tablePath.getTableName() + "' order by ordinal_position ";
    }

    @Override
    public List<String> bigDataColumns() {
        return List.of("BINARY", "VARBINARY", "IMAGE");
    }

    @Override
    public List<String> specialColumns() {
        return List.of("CURSOR", "GEOGRAPHY", "GEOMETRY", "HIERARCHYID", "JSON", "ROWVERSION", "SQL_VARIANT", "TABLE", "UNIQUEIDENTIFIER", "XML");
    }

    @Override
    public List<String> findFunctionIncompatible(String sql, String sourceCatalog, String sinkCatalog) {
        CharStream input = CharStreams.fromString(sql);

        // 创建词法分析器和解析器
        TSqlLexer lexer = new TSqlLexer(input);
        CommonTokenStream tokens = new CommonTokenStream(lexer);
        TSqlParser parser = new TSqlParser(tokens);

        // 从解析器中获取起始规则并进行解析
        ParseTree tree = parser.tsql_file();

        // 创建ParseTreeWalker
        ParseTreeWalker walker = new ParseTreeWalker();

        // 使用自定义Listener遍历解析树
        SqlServerFunctionListener listener = new SqlServerFunctionListener(sourceCatalog, sinkCatalog);
        walker.walk(listener, tree);
        return listener.getDifferentFunctions();
    }

    @Override
    public TableInfoModel getTableInfoByName(TablePath tablePath) throws CatalogException {
        TableInfoModel table = new TableInfoModel();
        try {
            String sql = "SELECT \n" +
                    "    t.name AS TABLE_NAME,\n" +
                    "    SUM(p.rows) AS TABLE_ROWS,\n" +
                    "    SUM(a.total_pages) * 8 AS TABLE_SIZE\n" +
                    "FROM \n" +
                    "    sys.tables t\n" +
                    "INNER JOIN      \n" +
                    "    sys.indexes i ON t.object_id = i.object_id\n" +
                    "INNER JOIN \n" +
                    "    sys.partitions p ON i.object_id = p.object_id AND i.index_id = p.index_id\n" +
                    "INNER JOIN \n" +
                    "    sys.allocation_units a ON p.partition_id = a.container_id\n" +
                    "INNER JOIN\n" +
                    "    sys.schemas s ON t.schema_id = s.schema_id\n" +
                    "WHERE \n" +
                    "    s.name = ? and t.name = ?\n" +
                    "GROUP BY \n" +
                    "    s.name, t.name;\n";
            log.info("Query mysql table info: {}", sql);
            PreparedStatement preparedStatement = getConnection(defaultUrl).prepareStatement(sql);
            preparedStatement.setString(1, tablePath.getSchemaName());
            preparedStatement.setString(2, tablePath.getTableName());
            ResultSet resultSet = preparedStatement.executeQuery();
            if (resultSet.next()) {
                table.setTableName(resultSet.getString("TABLE_NAME"));
                table.setRows(resultSet.getLong("TABLE_ROWS"));
                table.setComment(resultSet.getString("TABLE_COMMENT"));
                table.setSize(resultSet.getDouble("TABLE_SIZE"));
            }
            resultSet.close();
            preparedStatement.close();
        } catch (SQLException e) {
            log.error("获取表的信息失败！", e);
        }
        return table;
    }

    @Override
    public String getListSequenceSQL(String schema) {
        return String.format("SELECT \n" +
                "    seq.name AS SEQUENCE_NAME\n" +
                "FROM \n" +
                "    sys.sequences seq\n" +
                "INNER JOIN \n" +
                "    sys.schemas s ON seq.schema_id = s.schema_id\n" +
                "WHERE \n" +
                "    s.name = '%s'", schema);
    }

    @Override
    public String getCountDataSql(TablePath tablePath) {
        return "SELECT COUNT(*) FROM " + tablePath.getSchemaAndTableName();
    }

    @Override
    protected String getRandomPksSql(TablePath tablePath, String pk, int limit, int offset) {
        return String.format("SELECT TOP %s *\n" +
                "FROM %s\n" +
                "ORDER BY %s DESC", limit, getFullTableName(tablePath), pk);
    }

    @Override
    protected String getDataSqlNoPk(TablePath tablePath, int limit, int offset) {
        return String.format("SELECT TOP %s *\n" +
                "FROM %s\n" +
                "ORDER BY NEWID();", limit, getFullTableName(tablePath));
    }

    @Override
    public List<CharacteristicInfo> getCharacteristicInfo(String sql) {
        List<CharacteristicInfo> result = new ArrayList<>();

        CharStream input = CharStreams.fromString(sql);

        // 创建词法分析器和解析器
        TSqlLexer lexer = new TSqlLexer(input);
        CommonTokenStream tokens = new CommonTokenStream(lexer);
        TSqlParser parser = new TSqlParser(tokens);

        // 从解析器中获取起始规则并进行解析
        ParseTree tree = parser.tsql_file();

        // 创建ParseTreeWalker
        ParseTreeWalker walker = new ParseTreeWalker();

        // 使用自定义Listener遍历解析树
        SqlServerCharacteristicListener listener = new SqlServerCharacteristicListener();
        walker.walk(listener, tree);
        if (listener.isRowCount()) {
            result.add(new CharacteristicInfo("ROWCOUNT", "@@ROWCOUNT在目标数据库中不支持"));
        }
        if (listener.isXmlRef()) {
            result.add(new CharacteristicInfo("XML引用", "XML引用在目标数据库中不支持"));
        }

        return result;
    }

    @Override
    protected String getListIndexSQL(String defaultDatabase) {
        return String.format("SELECT \n" +
                "    sch.name AS SCHEMA_NAME,\n" +
                "    t.name AS TABLE_NAME,\n" +
                "    ind.name AS INDEX_NAME,\n" +
                "    ind.type_desc AS INDEX_TYPE,\n" +
                "    col.name AS COLUMN_NAME,\n" +
                "    ind.is_unique,\n" +
                "    ind.is_primary_key,\n" +
                "    ind.is_unique_constraint\n" +
                "FROM \n" +
                "    sys.indexes ind\n" +
                "INNER JOIN \n" +
                "    sys.tables t ON ind.object_id = t.object_id\n" +
                "INNER JOIN \n" +
                "    sys.schemas sch ON t.schema_id = sch.schema_id\n" +
                "INNER JOIN \n" +
                "    sys.index_columns ic ON ind.object_id = ic.object_id AND ind.index_id = ic.index_id\n" +
                "INNER JOIN \n" +
                "    sys.columns col ON t.object_id = col.object_id AND ic.column_id = col.column_id\n" +
                "WHERE \n" +
                "    sch.name = '%s'\n" +
                "ORDER BY \n" +
                "    TABLE_NAME, INDEX_NAME\n", defaultDatabase);
    }

    @Override
    public String[] getScriptDelimiters() {
        return new String[]{";", "GO"};
    }

    @Override
    public String[][] getBlockBoundStrings() {
        return TSQL_BEGIN_END_BLOCK;
    }
}
